package state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


/**
 * Create by zhh on 2022/9/15.
 * 不开启checkpoint：不重启
 *          * 开启checkpoint：
 *          * 1）如果没有配置重启策略：重启次数为Integer.MAX_VALUE
 *          * 2) 配置了重启策略，就使用配置的重启策略覆盖默认的
 *          *
 *          * 重启策略的配置：
 *          * 1)代码中
 *          * 2)yaml文件
 */
public class CheckPointAPP {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(5000);
        //设置StateBackend,将checkpoint保存到本地
        //env.setStateBackend(new FsStateBackend("file:///D:/idea/ideaProject/Flink/checkpoint"));

        //设置StateBackend,将checkpoint保存到hdfs
        //env.setStateBackend(new FsStateBackend("hdfs://192.168.32.222:9000/Flink/checkpoint"));

        //在job被挂掉或被取消掉时，保留checkpoint
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(//自定义设置重启策略
                        5,//重启次数
                        Time.seconds(5)//重启间隔
                        ));

        DataStreamSource<String> source = env.socketTextStream("node01", 9527);
        source.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                if (value.contains("zh")){
                    throw new RuntimeException("ZH coming..");
                }else {
                    return value.toUpperCase();
                }
            }
        })
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        String[] splits = value.split(",");
                        for (String split : splits) {
                            out.collect(split);
                        }
                    }
                }).map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value,1);
            }
        })
                .keyBy(x->x.f0)
                .sum(1)
                .print();
        env.execute("checkpointapp");
    }
}
